package com.lxn.app

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.google.gson.JsonObject
import com.lxn.bean.{PageActionLog, PageDisplayLog, PageLog, StartLog}
import com.lxn.until
import com.lxn.until.{MYKafkaUtils, MyOffsetUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.{SparkConf, rdd}
import org.apache.spark.streaming.Durations.seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}

object OdsBaseLogApp {

  def main(args: Array[String]): Unit = {

    //使用streamContext进行数据消费
    val sparkConf: SparkConf = new SparkConf().setAppName("ods_base_log_app").setMaster("local[4]")
    //获取streamContext对象
    val ssc = new StreamingContext(sparkConf,seconds(5))

    //现在开始消费数据
    //创建主题
    val topicName = "ODS_BASE_LOG"
    val topicGroup = "ODS_BASE_LOG_GROUP"

    //获取offset然后，根据指定的offset进行消费
    val offsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(topicGroup, topicName)
    //好了由于第一次消费，是没有偏移量的，所以需要进行默认消费
    var KfkDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(offsets!=null && offsets.nonEmpty){
      //存在偏移量，指定消费
      //无法识别重载可能是工具类里面的包存在冲突，导致使用了错误的类型
      KfkDStream = MYKafkaUtils.getConsumer(ssc,topicName,topicGroup,offsets)
    }
    else{
      //创建消费者组
      KfkDStream = MYKafkaUtils.getConsumer(ssc, topicName, topicGroup)
    }


    //由于现在存在数据重复消费或者数据漏消费问题，所以我们需要让它能够实现精确一次消费 可以使用事务，但是事务存在缺陷，或者操作复杂，所以这里我们使用偏移量和幂等性操作实现精确一次消费
    //那么，我们就需要将offsets提取出来，因为在后面已经转换结构了，所以我们不能以kafka的方式提交：一种是使用消费者提交，但是这里我们使用stream进行数据消费，没有消费者，其二是使用另一种类型提交，就是返回的那个类型
    //可是这两种方法都不适合我们使用，所以我们必须自己保存offstes进行提交，保证一次消费

    //这里我们不对数据进行处理，只是获取offsets
    var offsetRanges : Array[OffsetRange] = null
    val offsetRangeDStream :DStream[ConsumerRecord[String,String]] = KfkDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    //由于streamContext不能够直接获取数据，应为没有序列化，而kafka中有序列化操作，所以不需要我们序列化
    val JsonDStream: DStream[JSONObject] = offsetRangeDStream.map(
      ConsumerRecord => {
        //由于我们的数据是没有key的，所以我们值需要将value转化为可以传递的对象，这里我们将数据转化为json格式
        val log = ConsumerRecord.value()
        val jsonObject = JSON.parseObject(log)
        jsonObject
      }
    )
//    JsonDStream.print(100)
    //主题分流
    //我们需要将不同的数据看做不同的主题，然后发送到不同的为位置进行处理
    //查看数据，我们可以发现数据的行为不一样，根据这个分主题

    //启动主题
    val dwd_start_log : String = "DWD_START_LOG"
    //页面访问主题
    val dwd_page_log : String = "DWD_PAGE_LOG"
    //页面动作主题
    val dwd_page_action : String = "DWD_PAGE_ACTION"
    //页面曝光主题
    val dwd_page_display : String = "DWD_PAGE_DISPLAY"
    //错误主题
    val dwd_error_info : String ="DWD_ERROR_INFO"

    //先处理error数据，这种数据我们并没有很大意义，直接发往一个主题就好了
    //RDd数据流，需要用foreachRDD进行处理，对每一个rdd处理
    JsonDStream.foreachRDD(  //在rdd流里面遍历每一个rdd
      rdd => {     //这是单个rdd，包函一段时间产生的jsonObject对象
        rdd.foreachPartition(
          jsonObjItr => {
            for (jsonObject <- jsonObjItr) {
              val errJObject = jsonObject.getJSONObject("err") //如果是err，直接将他发送到一个主题
              if (errJObject != null) { //判断传过来的是不是err
                MYKafkaUtils.send(dwd_error_info, errJObject.toString) //生产者我们是在单列对象里面定义了一个producer
                // 所以我们可以直接使用工具类获得这个对象然后进行发送数据的操作
              } else {
                //提取公共信息
                //因为是json格式，所以有大JsonObject里面包含了小的JsonObject对象
                //分析数据，我们发现，数据中存在公共字段，对于不同的主题，我们传过去的数据也是不一样的，而这些公共字段其实也是需要传过去的，所以我们需要将他们凭借起来
                //获取公共字段的所有字段 获取小Json
                val commonObj: JSONObject = jsonObject.getJSONObject("common")
                val mid: String = commonObj.getString("mid")
                val uid: String = commonObj.getString("uid")
                val ar: String = commonObj.getString("ar")
                val ch: String = commonObj.getString("ch")
                val os: String = commonObj.getString("os")
                val md: String = commonObj.getString("md")
                val vc: String = commonObj.getString("vc")
                val isNew: String = commonObj.getString("is_new")
                //时间戳字段，它自己是一个小jsonObject,与common字段同级，所以我们只能用父json获取，但它又不是一个小json
                val ts: Long = jsonObject.getLong("ts")
                //分流页面日志  获取页面字段
                val pageObj: JSONObject = jsonObject.getJSONObject("page")
                if (pageObj != null) { //判断如果是一个page,就需要将它发送到一个地方
                  //提取字段  之后需要拼接公共字段
                  val pageId: String = pageObj.getString("page_id")
                  val pageItem: String = pageObj.getString("item")
                  val pageItemType: String = pageObj.getString("item_type")
                  val lastPageId: String = pageObj.getString("last_page_id")
                  val duringTime: Long = pageObj.getLong("during_time")
                  //封装 bean  拼接方式我们可以将他们封装成一个bean对象
                  val pageLog = PageLog(mid, uid, ar, ch, isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType, duringTime, ts)
                  //发送 kafka
                  //我们选择主题，和发送数据，这里要发送的数据是pageLog 的bean 但是由于是bean对象，所以不能直接转化为字符型，我们可以将它转化为jsonString类型，\
                  // 但是只能在java代码里面直接转换，因为java对象有set和get方法，可以根据字段获取值，而样例类里面没有这些方法，所以我们需要指定它直接使用字段，这样就OK了
                  MYKafkaUtils.send(dwd_page_log, JSON.toJSONString(pageLog, new SerializeConfig(true)))

                  //页面数据  分析数据，我们发现，这个曝光数据其实是一个jsonObject的数组，所以我们需要对这个数组里面，每一个对象进行处理
                  //首先使用父Json获取子json，这里是一个jsonObject数组
                  val displaysArr: JSONArray = jsonObject.getJSONArray("displays")
                  //好了，现在判断这个是不是为空，由于是数组，不能只判断是否为空，还需要判断数组里面是不是有值
                  if (displaysArr != null && displaysArr.size() > 0) {
                    //判断成立，现在该对数组里面每一个值进行处理，由于里面全是jsonObject，所以我们需要获取多个对象进行值处理
                    for (i <- 0 until displaysArr.size()) {
                      //好了，现在我们拿到了每一个jsonObject，我们现在只需要对每一个json对象进行封装，然后发送的哦消费者消费主题
                      val disJsonObt = displaysArr.getJSONObject(i)
                      //提取字段
                      val displayType: String = disJsonObt.getString("display_type")
                      val displayItem: String = disJsonObt.getString("item")
                      val displayItemType: String = disJsonObt.getString("item_type")
                      val displayOrder: String = disJsonObt.getString("order")
                      val displayPosId: String = disJsonObt.getString("pos_id")
                      //开始封装
                      val pageDisplayLog = PageDisplayLog(mid, uid, ar, ch,
                        isNew, md, os, vc, pageId, lastPageId, pageItem, pageItemType,
                        duringTime, displayType, displayItem, displayItemType, displayOrder, displayPosId, ts)
                      //发送数据到消费者
                      MYKafkaUtils.send(dwd_page_display, JSON.toJSONString(pageDisplayLog, new SerializeConfig(true)))

                    }

                  }

                  //提取时间
                  //查看数据，我们发现actions也是一个jsonObject数组
                  //还是先用大json获取下一级
                  val actionsArr = jsonObject.getJSONArray("actions")
                  //还是老套路，我直接写了
                  if (actionsArr != null && actionsArr.size() > 0) {
                    for (i <- 0 until actionsArr.size()) {
                      //获取数组中的每一个json对象
                      val actionsJObt = actionsArr.getJSONObject(i)
                      //提取字段
                      val actionId: String = actionsJObt.getString("action_id")
                      val actionItem: String = actionsJObt.getString("item")
                      val actionItemType: String = actionsJObt.getString("item_type")
                      //TODO actionts
                      val actionTs: Long = actionsJObt.getLong("ts")
                      //封装字段
                      val pageActionLog =
                        PageActionLog(mid, uid, ar, ch, isNew, md, os, vc,
                          pageId, lastPageId, pageItem, pageItemType,
                          duringTime, actionId, actionItem, actionItemType, actionTs)
                      //发送数据进行消费
                      MYKafkaUtils.send(dwd_page_action, JSON.toJSONString(pageActionLog, new SerializeConfig(true)))

                    }
                  }

                }
                //启动数据
                //这个start是和page一样的json对象，所以我们直接使用上级json对象获取start的json对象就可以了
                val startJObt = jsonObject.getJSONObject("start")
                //这只是个对象，判断是否为空就可以了
                if (startJObt != null) {
                  //不为空 提取字段
                  val entry: String = startJObt.getString("entry")
                  val loadingTimeMs: Long = startJObt.getLong("loading_time_ms")
                  val openAdId: String = startJObt.getString("open_ad_id")
                  val openAdMs: Long = startJObt.getLong("open_ad_ms")
                  val openAdSkipMs: Long = startJObt.getLong("open_ad_skip_ms")

                  //封装
                  val startLog =
                    StartLog(mid, uid, ar, ch, isNew, md, os, vc, entry, openAdId,
                      loadingTimeMs, openAdMs, openAdSkipMs, ts)
                  //发送数据
                  MYKafkaUtils.send(dwd_start_log, JSON.toJSONString(startLog, new SerializeConfig(true)))
                }
              }
            }
            // foreachPartition里面:  Executor段执行， 每批次每分区执行一次
            //刷写Kafka
            MYKafkaUtils.flush()
          }
        )

        //由于kafaka不会直接将数据放到磁盘中，而是将数据放入缓存区，但是这样可能会导致提交偏移量后数据并没有保存，所以我们需要强制刷写，
        //这样就需要换一个行动算子，对于每个分区进行刷写
        /*
        rdd.foreach(
          jsonObj => {
            //foreach里面:  提交offset???   A  executor执行, 每条数据执行一次.
            //foreach里面:  刷写kafka缓冲区??? executor执行, 每条数据执行一次.  相当于是同步发送消息.
          }
        )
         */

        //foreachRDD里面，forech外面: 提交offset??? B  Driver段执行，一批次执行一次（周期性）
        until.MyOffsetUtils.saveOffset(topicName,topicGroup,offsetRanges)
        //foreachRDD里面，forech外面: 刷写Kafka缓冲区??? B  Driver段执行，一批次执行一次（周期性） 分流是在executor端完成，driver端做刷写，刷的不是同一个对象的缓冲区.
      }
    )


    //开始项目
    ssc.start()
    ssc.awaitTermination()

  }


}
